package docsim;

import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

public class Reduce2 extends MapReduceBase implements
		Reducer<Text, Text, Text, Text> {

	private static final int ZERO = 0;
	private static final String COLON = ":";
	public static String dollar = "$";

	@Override
	public void reduce(Text key, Iterator<Text> values,
			OutputCollector<Text, Text> collector, Reporter arg3)
			throws IOException {
		StringBuffer sb = new StringBuffer();
		Iterator<Text> it = values;
		HashMap<String, Integer> map = new HashMap<String, Integer>();
		while (it.hasNext()) {
			String record[] = it.next().toString().split(COLON);
			String doc = record[ZERO];
			Integer freq = Integer.parseInt(record[1]);
			if (map.containsKey(doc)) {
				map.put(doc, map.get(doc) + freq);
			} else {
				map.put(doc, freq);
			}
		}
		boolean first = true;
		for (String doc : map.keySet()) {
			if (first) {
				sb.append(doc + COLON + map.get(doc));
				first = false;
			} else {
				sb.append(dollar).append(doc + COLON + map.get(doc));
			}
		}
		collector.collect(key, new Text(sb.toString()));
	}
}